import org.junit.Test;

import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

/**
 * 异步编程
 *git地址

 https://gitee.com/yujianlong1988/java8andDisruptor.git
 * @author yujianlong
 * @create 2018-06-18 21:33
 **/
public class FutrueTest {

	@Test
	public void FutrueTest1() {//创建
		CompletableFuture.runAsync(() -> System.out.println("asasa"));
		int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		//指定线程池
		ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_CORE_THREADS, 150, 20 * 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

		CompletableFuture<Void> asasa11 = CompletableFuture.runAsync(() -> System.out.println("asasa11"), executor);
		try {
			asasa11.get(5, TimeUnit.SECONDS);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		} catch (TimeoutException e) {
			e.printStackTrace();
		}

		//带返回值
		CompletableFuture<Integer> integerCompletableFuture = CompletableFuture.supplyAsync(() -> 1, executor);

		try {
			Integer result = integerCompletableFuture.get(5, TimeUnit.SECONDS);
			System.out.println(result);
		} catch (InterruptedException e) {
			e.printStackTrace();
		} catch (ExecutionException e) {
			e.printStackTrace();
		} catch (TimeoutException e) {
			e.printStackTrace();
		}

	}

	@Test
	public void FutrueTest2() {//同时执行多个，最终一起返回结果
		long start = System.currentTimeMillis();
		int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		//指定线程池
		ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_CORE_THREADS, 150, 20 * 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

		List<CompletableFuture<Integer>> fuList = IntStream.rangeClosed(1, 40).boxed().map(i -> CompletableFuture.supplyAsync(
				() -> {
					try {
						TimeUnit.SECONDS.sleep(1);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					return 1;
				}
				,executor)).collect(Collectors.toList());

		List<Integer> results = fuList.parallelStream().map(CompletableFuture::join).collect(Collectors.toList());
		long end = System.currentTimeMillis();
		System.out.println("耗时="+(end-start));
		System.out.println(results);
	}


	@Test
	public void FutrueTest3() {//有依赖的异步流水线
		long start = System.currentTimeMillis();
		int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		//指定线程池
		ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_CORE_THREADS, 150, 20 * 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

		List<CompletableFuture<Integer>> fuList = IntStream.rangeClosed(1, 40).boxed().map(i -> CompletableFuture.supplyAsync(
				() -> {
					try {
						TimeUnit.SECONDS.sleep(1);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					return 1;
				}
				, executor))
				.map(future -> future.thenApplyAsync(i -> i + 2,executor))
				.map(fu1 -> fu1.thenCompose(j -> CompletableFuture.supplyAsync(() -> j + 3,executor)))//异步流水线，会减少线程切换开销
				.collect(Collectors.toList());

		List<Integer> results = fuList.parallelStream().map(CompletableFuture::join).collect(Collectors.toList());
		long end = System.currentTimeMillis();
		System.out.println("耗时="+(end-start));
		System.out.println(results);
		System.out.println(results.size());
	}

	@Test
	public void FutrueTest4() {//无依赖的异步流水线
		long start = System.currentTimeMillis();
		int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		//指定线程池
		ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_CORE_THREADS, 150, 20 * 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

		//不相关的结合
		List<CompletableFuture<Integer>> fuList = IntStream.rangeClosed(1, 40).boxed().map(i -> CompletableFuture.supplyAsync(
				() -> {
					try {
						TimeUnit.SECONDS.sleep(1);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
					return 1;
				}
				, executor).thenCombine(CompletableFuture.supplyAsync(()->5),(i1,j1)->i1*j1)

				)

				.collect(Collectors.toList());

		List<Integer> results = fuList.parallelStream().map(CompletableFuture::join).collect(Collectors.toList());
		long end = System.currentTimeMillis();
		System.out.println("耗时="+(end-start));
		System.out.println(results);
		System.out.println(results.size());
	}



	@Test
	public void FutrueTest5() {//菱形结构(定时任务先后顺序模拟)  countdown也可以解决 disruptor尤其擅长

		//a---d-----|
		//b---------|------f
		//c---e-----|

		CompletableFuture<Void> a = CompletableFuture.runAsync(() -> System.out.println("a"));
		CompletableFuture<Void> d = a.thenRunAsync(()-> System.out.println("d"));

		CompletableFuture<Void> b = CompletableFuture.runAsync(() -> System.out.println("b"));

		CompletableFuture<Void> c = CompletableFuture.runAsync(() -> System.out.println("c"));
		CompletableFuture<Void> e = a.thenRunAsync(()-> System.out.println("e"));

		CompletableFuture<Void> f = CompletableFuture.runAsync(() -> {
			CompletableFuture.allOf(d,b,e);
			System.out.println("abcde准备好了");
		}).thenRunAsync(() -> System.out.println("f"));

		f.join();
	}
	@Test
	public void FutrueTest6() {//countdown实现菱形
		//a---d-----|
		//----b-----|------f
		//c---e-----|

		int DEFAULT_CORE_THREADS = Runtime.getRuntime().availableProcessors();
		//指定线程池
		ThreadPoolExecutor executor = new ThreadPoolExecutor(DEFAULT_CORE_THREADS, 150, 20 * 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(1024), new ThreadPoolExecutor.CallerRunsPolicy());

		CountDownLatch co1 = new CountDownLatch(3);
		CountDownLatch co2 = new CountDownLatch(1);
		CountDownLatch co3 = new CountDownLatch(1);
		Runnable a = () -> {
			System.out.println("a");
			co2.countDown();

		};
		Runnable d = () -> {
			System.out.println("d");
			co1.countDown();

		};
		Runnable b = () -> {
			System.out.println("b");
			co1.countDown();
		};
		Runnable c = () -> {
			System.out.println("c");
			co3.countDown();
		};
		Runnable e = () -> {
			System.out.println("e");
			co1.countDown();
		};

		Runnable f = () -> {
			System.out.println("f");
		};
		executor.submit(a);
		executor.submit(c);
		executor.submit(b);
		try {
			co2.await();
			co2.await();
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		executor.submit(d);
		executor.submit(e);
		try {
			co1.await();
			try {
				executor.submit(f).get();
			} catch (ExecutionException e1) {
				e1.printStackTrace();
			}
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
		executor.shutdown();
		try {
			executor.awaitTermination(10, TimeUnit.SECONDS);
		} catch (InterruptedException e1) {
			e1.printStackTrace();
		}
	}



}
